package com.ytktt.study.week12.task4.kafka.producer;

import com.ytktt.study.week12.task4.Order;
import com.ytktt.study.week12.task4.api.ProducerService;
import com.ytktt.study.week12.task4.properties.MQProperties;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;

/**
 * kafka 生产者有业务逻辑实现
 *
 * @author ytk
 */
@Slf4j
@RequiredArgsConstructor
public class KafkaProducerServiceImpl implements ProducerService {

    private final KafkaTemplate<String, Order> template;
    private final MQProperties mqProperties;

    @Override
    public void send(Order order) {
        MQProperties.FinancialKafkaProperties kafka = this.mqProperties.getKafka();
        int partition = Math.abs(order.getUserId().hashCode()) % kafka.getPartition();
        ProducerRecord<String, Order> record = new ProducerRecord<>(kafka.getTopic(), partition, order.getUserId(), order);
        ListenableFuture<SendResult<String, Order>> send = this.template.send(record);
        send.addCallback(success -> {
            RecordMetadata recordMetadata = success.getRecordMetadata();
            String topic = recordMetadata.topic();
            int writePartition = recordMetadata.partition();
            long offset = recordMetadata.offset();
            log.info("send order success，topic = {}, partition = {}, offset = {}", topic, writePartition, offset);
        }, failure -> {
            log.warn("send order fail：{}，{}", failure.getMessage(), failure);
            log.warn("save order {} in db", order);
        });
    }
}
